package com.bfxy.rabbit.producer.task;

import com.bfxy.rabbit.producer.broker.RabbitBroker;
import com.bfxy.rabbit.producer.enums.BrokerMessageStatusEnum;
import com.bfxy.rabbit.producer.entity.BrokerMessage;
import com.bfxy.rabbit.producer.service.MessageStoreService;
import com.bfxy.rabbit.task.annotation.ElasticJobConfig;
import com.dangdang.ddframe.job.api.ShardingContext;
import com.dangdang.ddframe.job.api.dataflow.DataflowJob;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.List;

@Slf4j
@Component
@ElasticJobConfig(
        name = "com.bfxy.rabbit.producer.task.RetryMessageDataflowJob",
        cron = "0/10 * * * * ?",
        description = "可靠性投递消息补偿任务",
        overwrite = true,
        shardingTotalCount = 1 // 这里因为测试只有一张表，但是真正生产环境可能有多张表，那这里的分片数就需要设置生多个，从这些表拉取数据，比如100张表，10个分片
)
public class RetryMessageDataflowJob implements DataflowJob<BrokerMessage> {

    /**
     * 最大重试次数
     */
    private static final int MAX_RETRY_COUNT = 3;

    @Resource
    private MessageStoreService messageStoreService;

    @Resource
    private RabbitBroker rabbitBroker;

    @Override
    public List<BrokerMessage> fetchData(ShardingContext shardingContext) {
        List<BrokerMessage> brokerMessageList = messageStoreService.fetchTimeOutMessage4Retry(BrokerMessageStatusEnum.SENDING);
        log.info("----------@@@@@@ 抓取数据集合，数量：{} @@@@@@----------", brokerMessageList.size());
        return brokerMessageList;
    }

    @Override
    public void processData(ShardingContext shardingContext, List<BrokerMessage> dataList) {
        dataList.forEach( brokerMessage -> {
            String messageId = brokerMessage.getMessageId();
            if (brokerMessage.getTryCount() >= MAX_RETRY_COUNT) {
                messageStoreService.updateStatusByMessageId(messageId, BrokerMessageStatusEnum.SEND_FAIL);
                log.warn("-----消息设置为最终失败，消息ID：{}-----", messageId);
            } else {
                // 重发逻辑
                // 每次重发的时候要更新下try_count字段
                messageStoreService.updateTryCountByMessageId(messageId);
                // 注意：无论是confirm消息还是reliant消息，发送消息以后broker都会去回调confirm。但是confirm消息不需要入库，而reliant消息需要，故消息重发使用reliant消息
                rabbitBroker.reliantSend(brokerMessage.getMessage());
            }
        });
    }

}
